package com.msunsoft.dataset.dictionary.validator;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.database.Database;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettlePluginException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.exception.KettleValueException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.core.row.value.ValueMetaFactory;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

import com.google.gdata.data.finance.CostBasis;
import com.msunsoft.dataset.dictionary.ecache.MapCache;

/**
 * Calculate new field values using pre-defined functions.
 * 
 * @author 许成利
 * @since 8-sep-2005
 */
public class Validator extends BaseStep implements StepInterface {
	private static Class<?> PKG = ValidatorMeta.class;

	private ValidatorMeta meta;

	private ValidatorData data;

	private List<Validation> validations = new LinkedList<Validation>();

	private MapCache cache = new MapCache();
	
	public Validator(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
			Trans trans) {
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
		meta = (ValidatorMeta) smi;
		data = (ValidatorData) sdi;
		Object[] r;

		if (first) {
			first = false;

			r = getRow(); // get row, set busy!
			if (r == null) { // no more input to be expected...

				setOutputDone();
				return false;
			}
			// 计算校验的字段个数
			data.fieldIndexes = new int[validations.size()];
			// 校验对象Validation key Set
			for (int i = 0; i < validations.size(); i++) {
				Validation validation = validations.get(i);
				if (!Const.isEmpty(validation.getFieldName())) {
					// 获取与校验对象相同字段名的输入流中字段位置
					if (getInputRowMeta().indexOfValue(validation.getFieldName())<0) {
						//throw new KettleStepException("在输入流中找不到要校验的字段[" + validation.getFieldName() + "]");//2015年12月1日09:50:29 找不到字段的不再抛出异常。将其移除list
						logBasic("在输入流中找不到要校验的字段[" + validation.getFieldName() + "]");
						validations.remove(i);
						i--;
					}else{
						data.fieldIndexes[i] = getInputRowMeta().indexOfValue(validation.getFieldName());
					}
				} else {
					throw new KettleStepException("步骤中校验器[" + validation.getFieldName() + "]没有指定校验字段!");
				}
			}
		} else {
			r = getRow(); // get row, set busy!
			if (r == null) { // no more input to be expected...

				setOutputDone();
				return false;
			}
		}

		if (log.isRowLevel()) {
			logRowlevel("Read row #" + getLinesRead() + " : " + getInputRowMeta().getString(r));
		}
		try {
			// 错误代码描述列，改造成中心字典代码e.getValidatorField().getDictionaryCode()
			// String
			// hospKeyCode="",hospValueAlias="",dictionaryCode="",dictionaryAlias="";

			List<KettleValidatorException> exceptions = validateFields(getInputRowMeta(), r);
			if (exceptions.size() > 0) {
				// TODO XUCL 2015年3月22日02:23:49 是否启用错误处理
				if (getStepMeta().isDoingErrorHandling()) {
					for (KettleValidatorException e : exceptions) {
						/*
						 * xucl 2015年5月14日10:38:51
						 * 克隆原有输入流InputRowMeta字段元数据生产新的元字段对象rowMeta，并增加中增加【医院值代码
						 * (hospKeyCode)】、
						 * 【医院值含义(hospValueAlias)】、【字典代码(dictionaryCode)】、【
						 * 字典代码含义(dictionaryAlias)】四个字段, 克隆数据流变量 r
						 * 生产新数据流newRowData，将新增四个字段值数据添加到数据流中
						 */
						RowMetaInterface rowMeta = getInputRowMeta().clone();
						int len = getInputRowMeta().size();

						Object[] newRowData = RowDataUtil.allocateRowData(rowMeta.size());
						if (newRowData != null) {
							System.arraycopy(r, 0, newRowData, 0, rowMeta.size());
						}
						ValueMetaInterface hospKeyCodeMeta = new ValueMeta("hospKeyCode".toUpperCase(),
								ValueMetaInterface.TYPE_STRING);
						hospKeyCodeMeta.setLength(100);
						rowMeta.addValueMeta(hospKeyCodeMeta);
						ValueMetaInterface hospValueAliasMeta = new ValueMeta("hospValueAlias".toUpperCase(),
								ValueMetaInterface.TYPE_STRING);
						hospValueAliasMeta.setLength(100);
						rowMeta.addValueMeta(hospValueAliasMeta);
						ValueMetaInterface dictionaryCodeMeta = new ValueMeta("dictionaryCode".toUpperCase(),
								ValueMetaInterface.TYPE_STRING);
						dictionaryCodeMeta.setLength(100);
						rowMeta.addValueMeta(dictionaryCodeMeta);
						ValueMetaInterface dictionaryAliasMeta = new ValueMeta("dictionaryAlias".toUpperCase(),
								ValueMetaInterface.TYPE_STRING);
						dictionaryAliasMeta.setLength(100);
						rowMeta.addValueMeta(dictionaryAliasMeta);
						newRowData[len] = e.getHospKeyCode();
						len++;
						newRowData[len] = e.getHospValueAlias();
						len++;
						newRowData[len] = e.getDictionaryCode();
						len++;
						newRowData[len] = e.getDictionaryAlias();
						/********************************* 增加四个字段结束 ***************************************/
						putError(rowMeta, newRowData, 1, e.getMessage(), e.getFieldname(),
								e.getValidatorField().getDictionaryCode() + "|"
										+ e.getValidatorField().getDictionaryAlias());
					}
				} else {
					KettleValidatorException e = exceptions.get(0);
					throw new KettleException(e.getMessage(), e);
				}
			} else {
				putRow(getInputRowMeta(), r); // copy row to possible alternate
												// rowset(s).
			}
		} catch (KettleValidatorException e) {
			if (getStepMeta().isDoingErrorHandling()) {
				// 错误代码描述列，改造成中心字典代码e.getValidatorField().getDictionaryCode()
				putError(getInputRowMeta(), r, 1, e.getMessage(), e.getFieldname(),
						e.getValidatorField().getDictionaryCode() + "|" + e.getValidatorField().getDictionaryAlias());
			} else {
				throw new KettleException(e.getMessage(), e);
			}
		}
		if (log.isRowLevel()) {
			logRowlevel("Wrote row #" + getLinesWritten() + " : " + getInputRowMeta().getString(r));
		}
		if (checkFeedback(getLinesRead())) {
			logBasic("Linenr " + getLinesRead());
		}

		return true;
	}

	/**
	 * @param inputRowMeta
	 *            the input row metadata
	 * @param r
	 *            the input row (data)
	 * @throws KettleValidatorException
	 *             in case there is a validation error, details are stored in
	 *             the exception.
	 */
	@SuppressWarnings("unchecked")
	private List<KettleValidatorException> validateFields(RowMetaInterface inputRowMeta, Object[] r)
			throws KettleValidatorException, KettleValueException {
		List<KettleValidatorException> exceptions = new ArrayList<KettleValidatorException>();
		// TODO校验对象Validation key Set
		for (int i = 0; i < validations.size(); i++) {
			Validation validation = validations.get(i);

			// TODO XUCL 2015年3月22日01:49:09 获取输入流中指定位置valueMeta,输入流中与校验字段相对应的字段值
			int valueIndex = data.fieldIndexes[i];
			ValueMetaInterface valueMeta = inputRowMeta.getValueMeta(valueIndex);
			String hospKeyCodeValue = valueMeta.getString(r[valueIndex]);// xuchengli
																			// 2015年9月28日23:22:52
																			// 将使用字典的字段值转换成sting类型
			boolean isNull = valueMeta.isNull(hospKeyCodeValue);
			
			if (!isNull && !Const.isEmpty(validation.getDictionaryCode())) {//字典转换
				// TODO 2015年9月28日23:16:51 拼接缓存key值
				StringBuffer strBuffer = new StringBuffer();
				if (!Const.isEmpty(hospKeyCodeValue)) {
					boolean found = false;
					//xuchengli 2015年11月30日13:20:10 公卫字典中单个字段存储有1,2,3这个格式的字典数据，需要拆分每个校验
					if(hospKeyCodeValue.contains(",")){
						String[] hospKeyCodeValueArray = hospKeyCodeValue.split(",");
						//多个字段校验，完全错误才是错误，增加校验次数
						int cycle = hospKeyCodeValueArray.length;
						int y = 0 ;
						StringBuffer tempBuffer = new StringBuffer();
						for (String keyCode : hospKeyCodeValueArray) {
							strBuffer.append(validation.getTableName()).append(".").append(validation.getDictionaryCode()).append(".").append(keyCode);
							String centerCodeValue = cache.getAsString(strBuffer.toString());
							if (!Const.isEmpty(centerCodeValue)) {
								found = true;
								tempBuffer.append(centerCodeValue).append(",");
							}
							strBuffer.delete(0, strBuffer.length());
						}//xuchengli 2015年11月30日13:20:10
						if(found){
							r[valueIndex] = tempBuffer.toString().subSequence(0, tempBuffer.length()-1);
						}
						if (!found) {
							KettleValidatorException exception = new KettleValidatorException(this, validation,
									KettleValidatorException.ERROR_VALUE_NOT_IN_LIST,
									validation.getTableName()+validation.getDictionaryCode()+hospKeyCodeValue+"==> not In DicList",
									validation.getFieldName(), valueMeta.getString(hospKeyCodeValue), null);
							exceptions.add(exception);
							r[valueIndex] = null;//字典校验异常数据，字段值复制为空
						}
					}else{
						strBuffer.append(validation.getTableName()).append(".").append(validation.getDictionaryCode()).append(".").append(hospKeyCodeValue);
						String centerCodeValue = cache.getAsString(strBuffer.toString());
						if (!Const.isEmpty(centerCodeValue)) {
							found = true;
							r[valueIndex] = centerCodeValue;
						}
						logDetailed("====>>读取待校验的字段值:" + hospKeyCodeValue + " 在缓存中查找key=[" + strBuffer.toString() + "]");
						if (!found) {
							KettleValidatorException exception = new KettleValidatorException(this, validation,
									KettleValidatorException.ERROR_VALUE_NOT_IN_LIST,
									strBuffer.toString()+"==> not In DicList",
									validation.getFieldName(), valueMeta.getString(hospKeyCodeValue), null);
							exceptions.add(exception);
							r[valueIndex] = null;//字典校验异常数据，字段值复制为空
						}
						strBuffer.delete(0, strBuffer.length());
					}
					
				}

			}
			//2015年12月1日10:49:43 xuchengli 增加非空校验
			if(isNull && validation.getIsMust()!=null && validation.getIsMust().equalsIgnoreCase("1")){
				KettleValidatorException exception = new KettleValidatorException(this, validation,
						KettleValidatorException.ERROR_NULL_VALUE_NOT_ALLOWED,
						validation.getTableName()+"."+validation.getFieldName()+"==> Null Value Not Allowed.",
						validation.getFieldName(), valueMeta.getString(hospKeyCodeValue), null);
				exceptions.add(exception);
			}
			
			
		}
		return exceptions;
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (ValidatorMeta) smi;
		data = (ValidatorData) sdi;
		if (super.init(smi, sdi)) {
			boolean passed = true;
			// TODO XUCL 2015年3月21日02:01:50 表结构与字典所在的数据库是否配置
			if (meta.getDatabaseMeta() == null) {
				logError("请配置字典值项的数据库信息!");
				passed = false;
			}

			// TODO XUCL 2015年3月21日02:01:57 步骤中用户配置的要进行校验的表名是否配置
			if (Const.isEmpty(meta.getTableName())) {
				logError(" 请填写校验名称!");
				passed = false;
			}
			if (!passed) {
				return false;
			}

			// TODO XUCL 2015年3月22日00:06:08 配置数据库信息并连接数据库
			data.db = new Database(this, meta.getDatabaseMeta());

			try {
				// 是否使用单独数据库连接
				if (getTransMeta().isUsingUniqueConnections()) {
					synchronized (getTrans()) {
						data.db.connect(getTrans().getTransactionId(), getPartitionID());
					}
				} else {
					data.db.connect(getPartitionID());
				}
				logBasic("数据集校验模块连接数据库...");

				// 初始化表中所有使用字典的列与字典值项
				// String sql = meta.getSQL();
				// TODO 2015年9月28日22:29:05 根据指定校验的数据表名称，查询出所有使用数据字典的字段
				// 2015年9月29日 16:37:08 徐晓东 增加sql语句
				// 2015年12月1日09:07:57 许成利  修改SQL语句支持must字段加载
				String sql = "SELECT T1.CUSTOMIDENTIFY AS TABLENAME,T.CUSTOMIDENTIFY  AS COLUMNNAME,T.DICTIONARYCODE  AS DICTIONARYCODE,T.MUST AS MUST FROM BD_DATASET_DATASETELEMENT T LEFT JOIN BD_DATASET_DATASETCATALOG T1 ON T1.CATALOGID = T.DATASETID WHERE T1.CUSTOMIDENTIFY = '"
						+ meta.getTableName() + "'";
				if (log.isDetailed()) {
					logDetailed("SQL query : " + sql);
				}
				ResultSet rs = data.db.openQuery(sql);
				if (rs == null) {
					logError("Couldn't open Query [" + sql + "]");
					setErrors(1);
					stopAll();
					return false;
				} else {
					// 初始化
					while (rs.next()) {
						String tableNameResult = rs.getString("tableName");
						String columnNameResult = rs.getString("columnName");
						String dictionaryCodeResult = rs.getString("dictionaryCode");
						String isMust = rs.getString("MUST");
						
						if (Const.isEmpty(tableNameResult)) {
							logError("初始化医院对照字典结果集中tableName字段有空置!");
							return false;
						}
						if (Const.isEmpty(columnNameResult)) {
							logError("初始化医院对照字典结果集中columnName字段有空置!");
							return false;
						}
						if (!Const.isEmpty(dictionaryCodeResult)) {//加载有字典代码的值域
							// TODO 2015年9月28日22:34:39
							// 根据dictionaryCodeResult查询字典对照表中数据
							String dicSQL = "SELECT T.DICTIONARYCODE AS DICTIONARYCODE,T.HISVALUE AS HISVALUE,T.CENTERVALUE AS CENTERVALUE FROM V_BD_DT_DIC_REG T WHERE T.DICTIONARYCODE='"
									+ dictionaryCodeResult + "'";
							ResultSet rs2 = data.db.openQuery(dicSQL);
							if (rs2 == null) {
								logError("Couldn't open Query [" + dicSQL + "]");
								setErrors(1);
								stopAll();
								return false;
							} else {
								while (rs2.next()) {
									String dictionaryCode = rs2.getString("dictionaryCode");
									String hisValue = rs2.getString("hisValue");
									String centerValue = rs2.getString("centerValue");

									// 生成字典值放入缓存中
									cache.put(tableNameResult+"."+dictionaryCode + "." + hisValue, centerValue);
									System.err.println(tableNameResult+"."+dictionaryCode + "." + hisValue);
								}
							}
							rs2.close();
						}
						
						
						// TODO 2015年9月28日22:34:39
						// 根据dictionaryCodeResult查询字典对照表中数据

						// TODO 2015年9月28日22:57:34 将表校验字段放置到list对象中
						Validation validation = new Validation();
						validation.setTableName(tableNameResult);
						validation.setFieldName(columnNameResult);
						validation.setDictionaryCode(dictionaryCodeResult);
						validation.setIsMust(isMust);//2015年12月1日09:10:51 xuchengli 增加非空校验
						validations.add(validation);
					}
					rs.close();
					logBasic("获取数据库字典数据完毕,并生成校验对象.");
				}
				return true;
			} catch (KettleException e) {
				logError("An error occurred, processing will be stopped: " + e.getMessage());
				setErrors(1);
				stopAll();
			} catch (SQLException sqle) {
				logError("An error occurred, processing will be stopped: " + sqle.getMessage());
				setErrors(1);
				stopAll();
			}
		}
		return true;
	}

	protected ValueMetaInterface createValueMeta(String name, int type) throws KettlePluginException {
		return ValueMetaFactory.createValueMeta(name, type);
	}

	protected ValueMetaInterface cloneValueMeta(ValueMetaInterface valueMeta, int type) throws KettlePluginException {
		return ValueMetaFactory.cloneValueMeta(valueMeta, type);
	}

}
